package org.example.kafka;

import com.alibaba.fastjson.support.odps.udf.CodecCheck;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.PartitionInfo;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.time.Duration;
import java.util.*;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;

@Slf4j
public class KafkaConsumerAnalysis {
    public static final String brokerList = "192.168.204.117:9092";
    public static final String topic = "topic-demo";
    public static final String groupId = "group.demo";

    public static final AtomicBoolean isRunning = new AtomicBoolean(true);

    public static final AtomicLong offset = new AtomicLong(-1l);

    public static Properties initConfig() {
        Properties properties = new Properties();
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, brokerList);
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);

        // 设置自动提交
        properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
        // 设置多久提交一次
        properties.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, 5000);
        return properties;
    }

    public static void main(String[] args) {
        Properties properties = initConfig();

        KafkaConsumer kafkaConsumer = new KafkaConsumer(properties);
//         kafkaConsumer.subscribe(Arrays.asList(topic));

        // 有哪些分区消息
        List<TopicPartition> partitions = new ArrayList<>();
        List<PartitionInfo> list = kafkaConsumer.partitionsFor(topic);
        list.forEach(item -> {
            partitions.add(new TopicPartition(item.topic(), item.partition()));
        });
        log.info(Arrays.toString(list.toArray(new PartitionInfo[]{})));
        kafkaConsumer.assign(partitions);
        try {
            while (isRunning.get()) {
                // 获取不到消息先阻塞
                // 从分区最开始点开始消费
                // ConsumerRecords records = kafkaConsumer.poll(Duration.ofHours(1));
                TopicPartition partition = new TopicPartition(topic, 0);
                // 如果从指定位置开始消费，保存分区信息，以及offset消息
                if (offset.get() == -1l) {
                    Map<TopicPartition, Long> map = kafkaConsumer.beginningOffsets(Arrays.asList(partition), Duration.ofMillis(10 * 1000));
                    map.forEach((topicPartion, offset) -> {
                        log.info(topicPartion.toString() + ", 位移 = " + offset);
                        kafkaConsumer.seek(topicPartion, offset);
                    });
                } else {
                    // 从指定位置开始消费
                    kafkaConsumer.seek(partition, offset.get() + 1);
                }
                ConsumerRecords records = kafkaConsumer.poll(Duration.ofMillis(10 * 1000));
                // seek之后怎么拿到数据
                Iterator iterator = records.iterator();
                while (iterator.hasNext()) {
                    ConsumerRecord<String, String> record = (ConsumerRecord<String, String>) iterator.next();
                    log.info("消息内容：" + record.value() + ", 所属分区：" + record.partition() + ", 所属位移：" + record.offset());
                    Thread.sleep(1000);
                    offset.set(record.offset());
                }
                // 消息处理完之后，手动同步提交
                kafkaConsumer.commitSync();
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            kafkaConsumer.close();
        }
    }
}
